package temporay;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;

/**
 * program: io-share
 * author: bsworld.xie
 * create: 2021-04-19 13:58
 * description:
 */
public class AcceptorThread extends AbstractSelectThread {

    private final ServerSocketChannel serverSocketChannel;
    private final SelectionKey acceptKey;
    private final Collection<SelectorThread> selectorThreads;
    private Iterator<SelectorThread> selectorIterator;

    public AcceptorThread(ServerSocketChannel ss, List<SelectorThread> selectorThreads) throws IOException {
        super("AcceptorThread:");
        this.serverSocketChannel = ss;
        this.acceptKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        this.selectorThreads = Collections.unmodifiableList(new ArrayList<SelectorThread>(selectorThreads));
        selectorIterator = this.selectorThreads.iterator();
    }

    public void run() {
        select();
    }

    private void select() {
        try {
            selector.select();

            Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selectedKeys.remove();

                if (!key.isValid()) {
                    continue;
                }
                if (key.isAcceptable()) {
                    if (!doAccept()) {
                        // If unable to pull a new connection off the accept
                        // queue, pause accepting to give us time to free
                        // up file descriptors and so the accept thread
                        // doesn't spin in a tight loop.
                        pauseAccept(10);
                    }
                }
            }
        } catch (IOException e) {
        }
    }

    private void pauseAccept(long millisecs) {
        acceptKey.interestOps(0);
        try {
            selector.select(millisecs);
        } catch (IOException e) {
            // ignore
        } finally {
            acceptKey.interestOps(SelectionKey.OP_ACCEPT);
        }
    }

    private boolean doAccept() {
        boolean accepted = false;
        SocketChannel sc = null;
        try {
            sc = serverSocketChannel.accept();
            accepted = true;
            sc.configureBlocking(false);

            // 通过轮询的方式将新建的连接分配到对应io线程中
            if (!selectorIterator.hasNext()) {
                selectorIterator = selectorThreads.iterator();
            }
            SelectorThread selectorThread = selectorIterator.next();
            boolean isSuccess = selectorThread.addAcceptedConnection(sc);
            if (!isSuccess) {
                throw new IOException("Unable to add connection to selector queue falied");
            }
        } catch (IOException e) {
        }
        return accepted;
    }

}
